package cn.smileyan.demo;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 *
 * @author Smileyan
 */
public class FlinkFlatMapDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Tuple2<String, Integer>> data = env.fromElements(
                Tuple2.of("hello", 1),
                Tuple2.of("world", 2));

        data.flatMap(new FlatMapFunction<Tuple2<String, Integer>, Tuple3<String, Integer, Long>>() {

            @Override
            public void flatMap(Tuple2<String, Integer> value, Collector<Tuple3<String, Integer, Long>> out) {
                String word = value.f0;
                int count = value.f1;
                out.collect(Tuple3.of(word, count, System.currentTimeMillis()));
            }
        }).print();

        env.execute("FlatMapFunction Example");
    }
}
